package com.gy.hsim.bservice.framework.mq;

import java.util.concurrent.ConcurrentLinkedQueue;

import com.gy.hsim.bservice.framework.bean.SystemConfig;
import com.gy.hsim.bservice.framework.spring.SpringUtil;


/**
 * 定义用于存储转发给MQ的消息队列
 * @author guiyi147
 *
 */
public class MQMessageQueue {
	
	private static MQMessageQueue mqMessageQueue;
	
	
	/**
	 * 存储转发到MQ 的消息队列
	 */
	protected  ConcurrentLinkedQueue<Object> mqEntityQueue = new ConcurrentLinkedQueue<Object>();

	private MQMessageQueue(MQParam mqParam){
		
		for (int i = 0; i < getTreadCount(mqParam); i++){
			new MQSendThread(new MQMessageSender(mqParam)).start();
		}
		
	}

	/**
	 * 返回单例对象
	 * @return
	 */
	public static MQMessageQueue instance(MQParam mqParam){
		if(mqMessageQueue == null) {
			mqMessageQueue = new MQMessageQueue(mqParam);
		}
		
		return mqMessageQueue;
	}
	
	/**
	 * 添加消息对象实体到队列中
	 * @param e
	 */
	public   void addEntityQueue(Object e){
		mqEntityQueue.add(e);
		
	}
	

	/**
	 * 发送消息的线程数
	 * @return
	 */
	public int getTreadCount(MQParam mqParam){
		return mqParam.getChannelCount() == 0 ? 2 : mqParam.getChannelCount();
	}
	
	/**
	 * 异步线程做消息发送，减少tigase发送消息到MQ的性能损失
	 *
	 */
	 class MQSendThread extends Thread{
		MessageSenderFactory mqSender = null;
		
		public MQSendThread(MessageSenderFactory mqSender){
			super.getName();
			this.mqSender = mqSender;
		}
		
		@Override
		public void run() {
			System.out.println("thread run()" +this.getId());
			Object entity = null;
			while(true){
				if( ( entity = mqEntityQueue.poll()) != null){
					System.out.println("thread" +this.getId());
					mqSender.send(entity);
					
				}else{
					try {
						sleep(1000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
				
			}
		}
		
		
	}

	
}
